package com.ztgf.message.api.client;

import com.alibaba.fastjson.JSON;
import com.ztgf.message.api.handler.ChannelRepository;
import com.ztgf.message.model.bo.MessageAnnouncementSaveBO;
import com.ztgf.message.model.constant.MessageTopicType;
import com.ztgf.message.model.dto.ResponseDTO;
import com.ztgf.message.model.service.IMessageAnnouncementService;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.RestController;

/**
 * author: chenyu.wang
 * date: 2020-05-13 14:23
 * Description:
 */
@RestController
public class KafkaReceiver {

    private ChannelRepository channelRepository = ChannelRepository.getInstance();

    private static Logger logger = LoggerFactory.getLogger(KafkaReceiver.class);

    @DubboReference
    private IMessageAnnouncementService messageAnnouncementService;

    /**
     * 监听系统具有具体消费者类型的消息
     * @param record
     */
    @KafkaListener(topics = {MessageTopicType.COMMENT_MESSAGE_TOPIC},errorHandler = "consumerAwareErrorHandler",groupId = "group-handle")
    public void listenSysConsumer(ConsumerRecord<?, ?> record) {
        Object value = record.value();
        if(null == record.value()){
            return;
        }
        ResponseDTO message = JSON.parseObject(value.toString(), ResponseDTO.class);
        logger.info("record : {}", record);
        logger.info("message : {}", message);

        MessageAnnouncementSaveBO messageAnnouncementSaveBO = JSON.parseObject(message.getData().toString(), MessageAnnouncementSaveBO.class);
        messageAnnouncementService.add(messageAnnouncementSaveBO);//保存
        getUnreadMessageNum(messageAnnouncementSaveBO.getShopId(),messageAnnouncementSaveBO.getShopType());
        //服务消息
        sendMessage(message);
    }

    /**
     * 获取未读消息数量
     * @param shopId
     * @param type 1买家 2卖家
     * @return
     */
    public Integer getUnreadMessageNum(Long shopId,int type){
        return messageAnnouncementService.updateUnreadMessageNum(shopId,type,0);
    }

    /**
     * netty发送消息
     * @param responseDTO
     */
    public void sendMessage(ResponseDTO responseDTO){
        channelRepository.sendMessage(responseDTO.getId() + "", responseDTO.getData());
    }
}
